import java.io.IOException;
import java.util.*;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.conf.*;
import org.apache.hadoop.io.*;
import org.apache.hadoop.mapred.*;
import org.apache.hadoop.util.*;

/**
 * Mapper for the Distributed Breadth Frist Search Fist, emmits the original
 * key value pair.  Then takes the distance to this key and the links out.
 * For each link, it emits <link, dist+1:> The fact that the link value pairs
 * have a null string for thier links is handled by the reducer.
 *
 * @author Ryan Hofler
 */
public class FinalMapper extends MapReduceBase
    implements Mapper<LongWritable, Text, IntWritable, Text> {

    public void map(LongWritable key,
		    Text value,
		    OutputCollector<IntWritable, Text> output,
		    Reporter reporter)
	throws IOException{

        String line = value.toString();
        String[] keyVal = line.split("\\t");
        String Key = keyVal[0];
        String sDist = keyVal[1];
        int Dist = Integer.parseInt(sDist);

        if(Dist != Integer.MAX_VALUE && !Key.equals(""))
            output.collect(new IntWritable(Dist), new Text(Key));

    }
}


